package org.zjt.spark.book

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.{SparkConf, SparkContext}

/**
  * 二次排序 + topN
  *
  */
object PageView extends App{
  var sparkConf = new SparkConf().setMaster("local[2]").setAppName("CommonFriends")
  val sc = new SparkContext(sparkConf)
  val broadcast:Broadcast[Int]= sc.broadcast(2)
  var array =sc.textFile("D:\\Idea workspace\\scala-demo\\src\\main\\resource\\pageview.txt").coalesce(2)
    .map(a=>{
      val data = a.split("   ")
      (data(0),data(1).toInt)
    }).reduceByKey(_+_).sortBy(_._2).mapPartitions(   //每个partition中获取广播变量，得到最大两个。总个数 = partition * 2(广播变量数)
    a =>{

      // mapPartitions 每个partition中对该分区的所有rdd进行操作

      val b = a.toStream
      val sort:Array[Int] = b.map(_._2).toArray.sorted
      val num = broadcast.value;
      b.filter( a => a._2 >= sort(num)).toIterator
    }).collect()


  //takeRight 得到后两个
  println(array.sortBy(_._2).takeRight(2).reverse.mkString("\n"))
  sc.stop()
}
